﻿#include "mqtt_handler.h"
#include "share_data.h"
#include "msg_type.h"
#include <thread>
#include <vector>
#include <map>
#include <fstream>
using namespace std;
using namespace std::placeholders;


MqttClient::MqttClient(const char* _ip, int _port, const char* _channel_name)
{
	handler_.server_ip = _ip;
	handler_.server_port = _port;
	handler_.channel_name = _channel_name;
	handler_.handler = this;
	//新建mqtt实例
	assert(_ip&&_port&&_channel_name);
	handler_.mosq = mosquitto_new(NULL, true, NULL);
	mosquitto_message_callback_set(handler_.mosq, &MqttHandler::callback);
	mosquitto_connect_callback_set(handler_.mosq, &MqttHandler::onconnect);
}
void MqttClient::Start()
{
	if (mosquitto_connect_async(handler_.mosq, handler_.server_ip.c_str(), handler_.server_port, 5) == MOSQ_ERR_SUCCESS)
	{
		LOGFMTD("%s %d",handler_.server_ip.c_str(),handler_.server_port);
		//std::thread th([=]() {mosquitto_loop_forever(handler_.mosq, 0, 1); });
		//th.detach();
		mosquitto_loop_start(handler_.mosq);
	}
	MqttHandler::GetInstance().Add_Client(this, handler_.channel_name.c_str());

}
void MqttClient::SubTopic(const char* _topic)
{
	mosquitto_subscribe(handler_.mosq, 0, _topic, 0);
	LOGFMTD("sub:%s",_topic);
	if (handler_.callback_map.find(_topic) == handler_.callback_map.end())
	{
		vector<mqtt_callback_child_t> tmp;
		tmp.push_back(std::bind(&MqttClient::default_callback,this,_1,_2,_3));
		handler_.callback_map.insert(make_pair(_topic, tmp));
	}
	else
	{
		handler_.callback_map[_topic].push_back(std::bind(&MqttClient::default_callback, this, _1, _2, _3));
	}
}
//默认数据回调接口
void MqttClient::default_callback(const char* _topic, const void* _pdata, int32_t _len)
{
	ShareData::GetInstance().MqttCallback( _topic, _pdata, _len);
}

void MqttClient::SubTopic(const char* _topic, mqtt_callback_child_t _func)
{
	mosquitto_subscribe(handler_.mosq, 0, _topic, 0);
	if (handler_.callback_map.find(_topic) == handler_.callback_map.end())
	{
		vector<mqtt_callback_child_t> tmp;
		tmp.push_back(_func);
		handler_.callback_map.insert(make_pair(_topic,tmp));
	}
	else
	{
		auto& tmp = handler_.callback_map[_topic];
		bool find = false;
		for (auto& v : tmp)
		{
			//!!!bug 无法有效比较成员函数指针。当类实例不同时，相同的成员函数指针依然相同
			if (v.target<void(*)>() == _func.target<void(*)>())
			{
				//find = true;
				//break;
			}
		}
		if (!find)
		{
			handler_.callback_map[_topic].push_back(_func);
		}
	}
}
void MqttClient::SubTopicWildcard(const char* _topic, mqtt_callback_child_t _func)
{
	string topic = _topic;
	if (topic.find("#") == string::npos && topic.find("+") == string::npos)
	{
		LOGFMTE("Not using Wildcard, try SubTopic function");
		return;
	}
	mosquitto_subscribe(handler_.mosq, 0, _topic, 0);
	LOGFMTD("sub:%s", _topic);
	if (handler_.special_callback_map.find(_topic) == handler_.special_callback_map.end())
	{
		vector<mqtt_callback_child_t> tmp;
		tmp.push_back(_func);
		handler_.special_callback_map.insert(make_pair(_topic, tmp));
	}
	else
	{
		auto& tmp = handler_.special_callback_map[_topic];
		bool find = false;
		for (auto v : tmp)
		{
			if (&v == &_func)
			{
				find = true;
				break;
			}
		}
		if (!find)
			handler_.special_callback_map[_topic].push_back(_func);
	}
}
void MqttClient::Publish(const char* _topic, const void* _payload, int32_t _len, uint8_t qos) 
{
	int ret = mosquitto_publish(handler_.mosq, 0, _topic, _len, _payload, qos, false);
}

map<std::string, mqtt_ins_t*> MqttHandler::map_;
map<std::string, MqttClient*> MqttHandler::ins_map_;
map<std::string, MqttClient*> MqttHandler::mqtt_map_;

MqttHandler::MqttHandler()
{
	mosquitto_lib_init();
}
void MqttHandler::Start()
{
	static bool start = false;
	if (start) return;
	start = true;
	auto identify = Setting::GetInstance().GetIdentify();
	//加载配置文件
	FILE* pfile = fopen("config/mqtt.json","rb");
	char* pbuf = new char[40960];
	if (pfile)
	{
		int len = fread(pbuf,1,40960,pfile);
		if (len > 0) pbuf[len] = 0;
		fclose(pfile);
	}
	//解析json配置文件
	nlohmann::json obj;
	try
	{
		obj = nlohmann::json::parse(pbuf);
	}
	catch (const std::exception& e)
	{
		LOGFMTE("mqtt.json parse fail %s",e.what());
		LOGFMTD("only using local mqtt 127.0.0.1:1883");
	}
	LOGFMTD("%s",obj.dump().c_str());
	delete[] pbuf;
	//创建本地mqtt
	auto local = new MqttClient("127.0.0.1", 1883, "local");
	//订阅消息,pb_report 上报数据
	local->SubTopic((identify + "/data/pb_report/local").c_str(), [=](const char* _topic, const void* _pdata, int32_t _len) {
		//pb_report 回调
		msgType::pb_report report;
		if (report.ParseFromArray(_pdata, _len))
			pb_report2data(report, "回放");
		});
	local->Start();
	int id = 0;
	for (auto item : obj)
	{
		if (item.contains("name") && item.contains("note") && item.contains("ip") && item.contains("port"))
		{
			using namespace msgType;
			auto channel_name = item["name"].get<std::string>();
			auto mqtt_ip = item["ip"].get<std::string>();
			auto mqtt_port = item["port"].get<int32_t>();
			auto channel_note = item["note"].get<std::string>();
			//搜索是否有重复的mqtt客户端
			auto key = mqtt_ip + ":" + std::to_string(mqtt_port);
			MqttClient* pclient = nullptr;
			if (mqtt_map_.find(key) == mqtt_map_.end())
			{
				pclient = new MqttClient(mqtt_ip.c_str(), mqtt_port, channel_name.c_str());
				pclient->SetNote(item["note"].get<std::string>().c_str());
				pclient->SetChannelId(id++);
				mqtt_map_.insert(std::make_pair(key,pclient));
				pclient->Start();
			}
			else
			{
				pclient = mqtt_map_[key];
				Add_Client(pclient,channel_name.c_str());
			}
			total_channel_.push_back(std::make_tuple(channel_note,channel_name));
		}
	}
}
//分解pb_report成多种小数据
void MqttHandler::pb_report2data(msgType::pb_report& _report,std::string _channel)
{						
	////加入优先级策略,记录每种通信设备的最后通信时间,3S无通信认为无效,可以切换下一优先级设备
	//int current_id = 0;
	//std::string channel_note = "WIFI";
	////获取当前通道id
	//for (int i = 0; i < total_channel_.size(); i++)
	//{
	//	if (get<1>(total_channel_[i]) == _channel)
	//	{
	//		current_id = i;
	//		channel_note = get<0>(total_channel_[i]);
	//		break;
	//	}
	//}
	////遍历totoal_channel,如果有小于当前通道的有效通道则直接返回
	//for (int i = 0; i < current_id; i++)
	//{
	//	//获取MQTTClient 指针
	//	auto mqtt_ptr = ins_map_.find(get<1>(total_channel_[i]));
	//	if (mqtt_ptr != ins_map_.end())
	//	{
	//		if (Setting::NowToMillseconds() - ins_map_[get<1>(total_channel_[i])]->last_recv_stamp_ < 3000 && ins_map_[get<1>(total_channel_[i])]->last_recv_stamp_ != -1)
	//			return;
	//	}
	//}

	using namespace msgType;
	auto report = GET_SHARE_DATA(pb_report);
	GET_SHARE_DATA_CONTAINER(pb_report)->UpdateStamp();
	report->CopyFrom(_report);

	//电池数据
	auto bat_item = _report.bat();
	for (int i = 0; i < bat_item.bat_size(); i++)
	{
		auto bat_ptr = GET_SHARE_DATA(pb_bat,bat_item.bat(i).name());
		GET_SHARE_DATA_CONTAINER(pb_bat, bat_item.bat(i).name())->UpdateStamp();
		bat_ptr->CopyFrom(bat_item.bat(i));
	}
	//gps数据
	auto gps_item = _report.gps();
	for (int i = 0; i < gps_item.gps_size(); i++)
	{
		auto gps_ptr = GET_SHARE_DATA(pb_gps, gps_item.gps(i).name());
		GET_SHARE_DATA_CONTAINER(pb_gps, gps_item.gps(i).name())->UpdateStamp();
		gps_ptr->CopyFrom(gps_item.gps(i));
	}
	//高度计数据
	auto altimeter_item = _report.altimeter();
	for (int i = 0; i < altimeter_item.altimeter_size(); i++)
	{
		auto altimeter_ptr = GET_SHARE_DATA(pb_altimeter, altimeter_item.altimeter(i).name());
		GET_SHARE_DATA_CONTAINER(pb_altimeter, altimeter_item.altimeter(i).name())->UpdateStamp();
		altimeter_ptr->CopyFrom(altimeter_item.altimeter(i));
	}
	//bms数据
	auto bms = _report.bms();
	{
		auto bms_ptr = GET_SHARE_DATA(pb_bms);
		GET_SHARE_DATA_CONTAINER(pb_bms)->UpdateStamp();
		bms_ptr->CopyFrom(bms);
	}
	//电机数据
	auto motor_item = _report.motor();
	for (int i = 0; i < motor_item.motor_size(); i++)
	{
		auto motor_ptr = GET_SHARE_DATA(pb_motor, motor_item.motor(i).name());
		GET_SHARE_DATA_CONTAINER(pb_motor, motor_item.motor(i).name())->UpdateStamp();
		motor_ptr->CopyFrom(motor_item.motor(i));
	}
	//压力传感器数
	auto press_item = _report.pressure();
	for (int i = 0; i < press_item.press_size(); i++)
	{
		auto press_ptr = GET_SHARE_DATA(pb_pressure, press_item.press(i).name());
		GET_SHARE_DATA_CONTAINER(pb_pressure, press_item.press(i).name())->UpdateStamp();
		press_ptr->CopyFrom(press_item.press(i));
	}
	//惯导数据
	auto ins = _report.ins();
	{
		auto ins_ptr = GET_SHARE_DATA(pb_ins);
		GET_SHARE_DATA_CONTAINER(pb_ins)->UpdateStamp();
		ins_ptr->CopyFrom(ins);
	}
	//舵机数据
	auto rudder_item = _report.rudder2();
	for (int i = 0; i < rudder_item.rudder_size(); i++)
	{
		auto rudder_ptr = GET_SHARE_DATA(pb_rudder, rudder_item.rudder(i).name());
		GET_SHARE_DATA_CONTAINER(pb_rudder, rudder_item.rudder(i).name())->UpdateStamp();
		rudder_ptr->CopyFrom(rudder_item.rudder(i));
	}
	//任务数据
	auto mis = _report.mission();
	{
		auto mis_ptr = GET_SHARE_DATA(pb_mission);
		GET_SHARE_DATA_CONTAINER(pb_mission)->UpdateStamp();
		mis_ptr->CopyFrom(mis);
	}
	//dvl数据
	auto dvl = _report.dvl();
	{
		auto dvl_ptr = GET_SHARE_DATA(pb_dvl);
		GET_SHARE_DATA_CONTAINER(pb_dvl)->UpdateStamp();
		dvl_ptr->CopyFrom(dvl);
	}
	//温湿度数据
	{
		for (int i = 0; i < _report.bme280_size(); i++)
		{
			auto bme_ptr = GET_SHARE_DATA(pb_bme280, _report.bme280(i).name());
			GET_SHARE_DATA_CONTAINER(pb_bme280, _report.bme280(i).name())->UpdateStamp();
			bme_ptr->CopyFrom(_report.bme280(i));
		}
	}
	//异常数据
	auto err = _report.err2();
	{
		auto err_ptr = GET_SHARE_DATA(pb_error2);
		GET_SHARE_DATA_CONTAINER(pb_error2)->UpdateStamp();
		err_ptr->CopyFrom(err);
	}
	//更新通信通道
	auto ch_ptr = GET_SHARE_DATA(pb_communicate, "src");
	//GET_SHARE_DATA_CONTAINER(pb_communicate, "src")->UpdateStamp();
	ch_ptr->set_channel(_channel);
	ch_ptr->set_name("src");
	ch_ptr->set_signal(ch_ptr->signal() + 1);
	ch_ptr->set_r_stamp(Setting::NowToMillseconds());

}
MqttHandler::~MqttHandler()
{
	mosquitto_lib_cleanup();
}
//处理订阅
void MqttHandler::onconnect(struct mosquitto* mosq, void* data, int result)
{
	if (!result)
	{
		if (map_.find(to_string((int64_t)mosq)) != map_.end())
		{
			auto cmap = map_[to_string((int64_t)mosq)]->callback_map;
			for(auto it=cmap.begin();it!=cmap.end();it++)
			{
				mosquitto_subscribe(mosq,0,it->first.c_str(),0);
			}
			auto smap = map_[to_string((int64_t)mosq)]->special_callback_map;
			for (auto it = smap.begin(); it != smap.end(); it++)
			{
				mosquitto_subscribe(mosq, 0, it->first.c_str(), 0);
			}
		}
	}
}
//经过这一级转换，回调由static转为非static函数
void MqttHandler::send_sig(const char* _topic, const void* _pdata, int32_t _len, struct mosquitto* _mosq)
{
	if (map_.find(to_string((int64_t)_mosq)) == map_.end()) return;
	auto cmap = map_[to_string((int64_t)_mosq)]->callback_map;
	if (cmap.find(_topic) != cmap.end())
	{
		for (auto v : cmap[_topic])
		{
			v(_topic, _pdata, _len);
		}
	}
	//special map,通配符处理; 逐级处理
	auto smap = map_[to_string((int64_t)_mosq)]->special_callback_map;
	for (auto it = smap.begin(); it != smap.end(); it++)
	{
		auto topic_result = SplitString(std::string(_topic), "/");
		auto sub_result = SplitString(it->first, "/");
		bool match_flag = true;
		for (int i = 0; i < sub_result.size(); i++)
		{
			if (sub_result[i] == "#")//最后一级,mqtt只能最后一级是#号
			{
				break;
			}
			else if (sub_result[i] == "+")
			{
				continue;
			}
			else
			{
				if (topic_result.size() > i)
				{
					if (topic_result[i] != sub_result[i])
					{
						match_flag = false;
						break;
					}
				}
			}
		}
		if (match_flag)
		{
			for (auto v : it->second)
			{
				v(_topic, _pdata, _len);
			}
		}
	}
	/*try
	{
		uint8_t* ptr = new uint8_t[_len];
		memcpy(ptr,_pdata,_len);
		auto data = std::make_tuple(_topic,ptr,_len,(int64_t)_mosq);
		mqtt_data_que_.Enqueue(data);
	}
	catch (const std::exception&)
	{

	}*/
}
//优化函数，防止函数阻塞
void MqttHandler::callback(mosquitto* _mosq, void*, const mosquitto_message* _pdata)
{
	MqttHandler::GetInstance().send_sig(_pdata->topic,_pdata->payload,_pdata->payloadlen,_mosq);
}
vector<tuple<string,string>> MqttHandler::GetTotalChannel()
{
	return total_channel_;
}
void MqttHandler::Sub(const char* _topic)
{
	regex re("data/pb_.*");
	if ( !regex_match(_topic, re))
	{
		LOGFMTE("format error. only [.*/data/pb_.*]");
		return;
	}
	using namespace msgType;
	using namespace std::placeholders;
	auto identify = Setting::GetInstance().GetIdentify();
	if ((*this)["local"])
	{
		(*this)["local"]->SubTopic((identify + "/" + _topic).c_str(),\
			std::bind(&ShareData::MqttCallback,&ShareData::GetInstance(),_1,_2,_3));
	}
	else
	{
		LOGFMTE("local mqtt not create");
	}
}
vector<string> MqttHandler::SplitString(const string& str, const string& delim)
{
	vector<string> tmp;
	if (str.length() == 0) return tmp;
	if (delim.length() == 0) return tmp;
	for (unsigned char c : str)
	{
		if (c > 127)
			return tmp;
	}
	for (unsigned char c : delim)
	{
		if (c > 127)
			return tmp;
	}
	string detail = str;
	//if source not contail delim return size 0;
	if (detail.find(delim) == string::npos)
	{
		tmp.push_back(str);
		return tmp;
	}
	//if delim located at the end of source ,del it
	if (detail.find_last_of(delim) + delim.length() == detail.length())
		detail = detail.substr(0, detail.length() - delim.length());
	string::size_type pos = string::npos;
	while ((pos = detail.find(delim)) != string::npos)
	{
		if (pos != 0)
		{
			tmp.push_back(detail.substr(0, pos));
			detail = detail.substr(pos + delim.length());
		}
		else
		{
			detail = detail.substr(1);
			tmp.push_back("");
		}
	}
	tmp.push_back(detail);
	return tmp;
}